package com.jsean.core.consumer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import net.openhft.chronicle.queue.ChronicleQueue;
import net.openhft.chronicle.queue.ExcerptTailer;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

public class LogConsumer {
    private static final Logger logger = LoggerFactory.getLogger(LogConsumer.class);

    private final String consumerId;

    public LogConsumer(String consumerId) {
        this.consumerId = consumerId;
    }

    public void start() {
        String path = "store";

        try (ChronicleQueue queue = ChronicleQueue.singleBuilder(path).build()) {
            ExcerptTailer tailer = queue.createTailer();

            long lastReadIndex = OffsetStore.getOffset(consumerId);
            if (lastReadIndex > 0) {
                tailer.moveToIndex(lastReadIndex);
                logger.info("{}: Resuming from index {}", consumerId, lastReadIndex);
            }

            while (true) {
                boolean read = tailer.readDocument(wire -> {
                    String msg = wire.read(() -> "msg").text();
                    logger.debug("{} received: {}", consumerId, msg);
                });

                if (read) {
                    long currentIndex = tailer.index();
                    OffsetStore.setOffset(consumerId, currentIndex);
                } else {
                    Thread.sleep(100); // 等待新数据
                }
            }
        } catch (Exception e) {
            logger.error("{} error: {}", consumerId, e.getMessage(), e);
        }
    }

    public static void main(String[] args) {
        Properties props = new Properties();
        try (InputStream is = LogConsumer.class.getClassLoader().getResourceAsStream("application.properties")) {
            props.load(is);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        String consumerId = props.getProperty("app.consumer.id");
        new LogConsumer(consumerId).start();
    }
}